package com.zhao.demo.unbound.demo01_wordcount

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * Description: 无界流演示(使用flink无界流处理实时监控netcat客户端和服务器交互的数据,进行实时的计算,并将计算结果展示出来)<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/11/24 15:26
 *
 * @author 柒柒
 * @version : 1.0
 */

object UnboundedFlowDemo {
  def main(args: Array[String]): Unit = {
    //1.拦截非法的参数
    if (args == null || args.length != 4){
      println(
        """
          |警告！录入参数不符合要求！
          |""".stripMargin)
      sys.exit(-1)
    }

    //2.获得待计算的源的path,以及计算后的目的地的path
    val tool: ParameterTool = ParameterTool.fromArgs(args)

    val hostname = tool.get("hostname")
    val port = tool.getInt("port")

    //3.执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //设置所有算子的全局并行度
    //env.setParallelism(1)

    //4.计算,显示或者保存结果

    //a)导入单例类scala中隐式成员
    import org.apache.flink.api.scala._

    //b)迭代计算
    env.socketTextStream(hostname,port)
      .flatMap(_.split("\\s+"))
      .filter(_.nonEmpty)
      .map((_,1))
      .keyBy(0)
      .sum(1)
      .print()

    //5.启动
    env.execute(this.getClass.getSimpleName)
  }
}
